package com.sqoop.server;

import com.sqoop.core.msg.Msg;
import com.sqoop.server.factory.TaskManager;
import com.sqoop.server.state.StatInfo;
import io.netty.channel.udt.nio.NioUdtAcceptorChannel;
import org.apache.commons.lang3.StringUtils;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;


/**
 * Created by - on 2017/11/9.
 */
public abstract class BaseProcess {
    private static Logger logger = LogManager.getLogger(BaseProcess.class.getName());

    private String taskId;
    private ThreadPoolExecutor threadPool;
    private int msgSize = 1000;
    private int threadSize = 10;
    private BlockingQueue<Runnable> threadBlockingQueue = new ArrayBlockingQueue<>(5000);
    private BlockingQueue<Msg> msgBlockingQueue = new ArrayBlockingQueue<>(5000);
    private volatile boolean suspendFlag = true;
    private List<WorkerThread> threadList = new ArrayList<>();

    private Map<String,AtomicLong> threadstatsuccessmap = new ConcurrentHashMap<>();
    private Map<String,AtomicLong> threadstatfailedmap = new ConcurrentHashMap<>();

    public abstract void process(Msg msg);
    public abstract void init();

    public abstract void stop0();
    public abstract void pause0();


    private AtomicInteger receiveSize = new AtomicInteger(0);
    private AtomicInteger processSize = new AtomicInteger(0);
    private AtomicInteger errorSize = new AtomicInteger(0);
    private long startTime = System.currentTimeMillis();


    public void addError(){
        errorSize.incrementAndGet();
    }

    public void addProcess(){
        processSize.incrementAndGet();
    }

    public void addReceive(){
        receiveSize.incrementAndGet();
    }


    public StatInfo stat(){
        StatInfo statInfo = new StatInfo();
        statInfo.setErrorsSize(errorSize.get());
        statInfo.setProcessSize(processSize.get());
        statInfo.setReceiveSize(receiveSize.get());
        statInfo.setRunningTime(System.currentTimeMillis() - startTime);
        return statInfo;
    }

    public void aSend(Msg msg){

        msg.setSrcTaskName(this.taskId);

        String dstTaskName = msg.getDstTaskName();
        if (StringUtils.isEmpty(dstTaskName)){
            processNoDstRouterTask(msg);
            logger.error("receive a msg ,dstTaskName is null!!!!" );
            return;
        }
        if (TaskManager.getInstance().getBaseProcessMap().containsKey(msg.getDstTaskName())){
            TaskManager.getInstance().getBaseProcessMap().get(dstTaskName).addMsg(msg);
        }else {
            logger.error("receive a msg ,dstTaskName is {},but there is no task in taskmanager" ,dstTaskName);
        }

    }

    private   void processNoDstRouterTask(Msg msg){

    }


    public void start(){
        init();
        threadPool = new ThreadPoolExecutor(threadSize, threadSize, 10, TimeUnit.MINUTES, threadBlockingQueue);
        for (int i = 0; i < threadSize; i++){
            WorkerThread thread = new WorkerThread(this);
            thread.setName(taskId + "_i");
            thread.setMsgId(this.taskId);
            threadPool.execute(thread);
            threadList.add(thread);
        }
    }

    public void stop() {

        Runnable runnable = new Runnable() {
            @Override
            public void run() {
                stop0();
            }
        };
        Thread stop = new Thread(runnable);
        try {
            stop.join(10000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        msgBlockingQueue = new ArrayBlockingQueue<>(0);
        threadPool.shutdownNow();
    }


    public void pause(){
        pause0();
        for (WorkerThread thread : threadList) {
            thread.setSuspend(true);
        }
    }


    public void resume(){

        for (WorkerThread thread : threadList) {
            thread.setSuspend(false);
        }
    }



    public void addMsg(Msg msg){
        if (msg == null){
            logger.error("receive a null msg!!");
            return;
        }
        this.msgBlockingQueue.offer(msg);
    }

    public BlockingQueue<Runnable> getThreadBlockingQueue() {
        return threadBlockingQueue;
    }

    public void setThreadBlockingQueue(BlockingQueue<Runnable> threadBlockingQueue) {
        this.threadBlockingQueue = threadBlockingQueue;
    }

    public BlockingQueue<Msg> getMsgBlockingQueue() {
        return msgBlockingQueue;
    }

    public void setMsgBlockingQueue(BlockingQueue<Msg> msgBlockingQueue) {
        this.msgBlockingQueue = msgBlockingQueue;
    }


    public Map<String, AtomicLong> getThreadstatsuccessmap() {
        return threadstatsuccessmap;
    }

    public void setThreadstatsuccessmap(Map<String, AtomicLong> threadstatsuccessmap) {
        this.threadstatsuccessmap = threadstatsuccessmap;
    }

    public Map<String, AtomicLong> getThreadstatfailedmap() {
        return threadstatfailedmap;
    }

    public void setThreadstatfailedmap(Map<String, AtomicLong> threadstatfailedmap) {
        this.threadstatfailedmap = threadstatfailedmap;
    }

    public void setTaskId(String taskId) {
        this.taskId = taskId;
    }

    public String getTaskId() {
        return taskId;
    }
}